// Copyright 2013 Tumblr, Inc.
// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package client

import (
	"sort"
	"sync"

	"vena"
	"vena/server"

	"circuit/kit/sched/limiter"
	"circuit/use/e"
)

func (cli *Client) Query(q *server.Query) (result server.Series, err error) {
	cli.qlmtr.Open()
	defer cli.qlmtr.Close()

	// Catch cross-call panics
	defer func() {
		if r := recover(); r != nil {
			err = e.NewError("Vena server QUERY call panic:\n%#v\n", r)
		}
	}()

	cli.lk.Lock()
	shs := cli.metric.Dump()
	cli.lk.Unlock()

	var lk sync.Mutex
	bin := make(map[vena.Time]*server.Sketch)
	lmtr := limiter.New(len(shs))
	for _, sh := range shs {
		x := sh.(*shard).X
		lmtr.Go(func() {
			retrn := x.Call("Query", q)
			lk.Lock()
			defer lk.Unlock()
			if retrn[1] != nil {
				err = retrn[1].(error)
				return
			}
			for _, s := range retrn[0].(server.Series) {
				skt, present := bin[s.Time]
				if !present {
					skt = &server.Sketch{Time: s.Time}
					bin[s.Time] = skt
				}
				skt.Sum += s.Sum
				skt.Weight += s.Weight
			}
		})
	}
	lmtr.Wait()
	if err != nil {
		return nil, err
	}

	result = make(server.Series, 0, len(bin))
	for _, skt := range bin {
		result = append(result, skt)
	}
	sort.Sort(result)

	return result, nil
}
